package cn.monkey.transport.netty;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import reactor.core.publisher.Flux;

import java.util.function.Predicate;

public class SimpleNettyConnection implements NettyConnection {

    protected final Channel channel;

    protected final Flux<Object> flux;

    public SimpleNettyConnection(Channel channel) {
        this.channel = channel;
        this.flux = Flux.create(sink -> channel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                sink.next(msg);
                super.channelRead(ctx, msg);
            }
        }));
    }

    @Override
    public Channel channel() {
        return this.channel;
    }

    @Override
    public <T> Flux<T> receive(Class<T> clazz, Predicate<T> p) {
        return this.flux.filter(clazz::isInstance)
                .map(clazz::cast)
                .filter(p);
    }
}
